package com.gitee.l0km.xthrift.mq;

import gu.simplemq.Channel;
import gu.simplemq.IMessageAdapter;
import gu.simplemq.IMessageQueueFactory;
import gu.simplemq.IPublisher;
import gu.simplemq.ISubscriber;
import gu.simplemq.SimpleLog;
import gu.simplemq.exceptions.SmqUnsubscribeException;

import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.MoreObjects.firstNonNull;
import static com.google.common.base.Strings.isNullOrEmpty;

import java.util.Collections;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import javax.annotation.PostConstruct;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.gitee.l0km.com4j.basex.DefaultExecutorProvider;
import com.gitee.l0km.xthrift.mq.exception.InvalidResponseException;
import com.gitee.l0km.xthrift.mq.exception.InvocationRuntimeException;
import com.gitee.l0km.xthrift.mq.exception.TApplicationException;
import com.gitee.l0km.xthrift.mq.exception.ResponseException;
import com.google.common.base.Supplier;
import com.google.common.base.Suppliers;
import com.google.common.base.Throwables;
import com.google.common.primitives.Primitives;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import com.google.common.util.concurrent.UncheckedExecutionException;

/**
 * 基于消息系统实现远程调用(RPC)
 * @author guyadong
 * @since 1.5.0
 */
@Component
public class MqInvoker implements IMessageAdapter<Envelope>{
	/** 调用成功数据的字段名 */
	private static final String FIELD_SUCCESS = "success";
	/** 运行时异常数据的字段名 */
	private static final String FIELD_EXCEPTION = "exception";
	/** org.apache.thrift.TApplicationException 异常的数据字段名 */
	private static final String FIELD_MESSAGE = "message";
	/** org.apache.thrift.TApplicationException 异常数据中的类型字段名 */
	private static final String FIELD_TYPE = "type";
	private static long DEFAULT_TASK_TIMEOUT_MILLIS = TimeUnit.SECONDS.toMillis(20);
	/** 负责为每一个请求分配唯一的序列号，确保并发环境下，返回的响应数据分发到正确的 {@link ListenableFuture}  */
	private static final AtomicInteger SEQID = new AtomicInteger();
	/** 以请求序列号为KEY 保存当前所有请求的调用上下文对象 */
	private static final Map<Integer, InvocationContext<?>> invocationContexts = new ConcurrentHashMap<>();
	private final IPublisher publisher;
	private final ISubscriber subscriber;
	private final long taskTimeoutMillis;
	private final ScheduledExecutorService scheduledExecutor;
	private final Class<? extends Exception> exceptionType;
	private final Supplier<String> ackSupplier;
	private volatile String autoAck;
	
	public MqInvoker(@Autowired Builder builder) {
		checkNotNull(builder.messageQueueFactory, "messageQueueFactory is null");
		this.publisher = builder.messageQueueFactory.getPublisher();
		this.subscriber = builder.messageQueueFactory.getSubscriber();
		this.taskTimeoutMillis = builder.taskTimeoutMillis;
		this.scheduledExecutor = firstNonNull(builder.scheduledExecutor,DefaultExecutorProvider.getTimerExecutor());
		this.exceptionType = firstNonNull(builder.exceptionType, InvocationRuntimeException.class);
		this.ackSupplier = builder.ackSupplier;
	}
	/** 启动客户端订阅ACK频道 */
	@PostConstruct
	public void start() {
		subscriber.register(new Channel<>(getAckChannel(), Envelope.class,this));
		SimpleLog.log("Register ACK Channel {}",getAckChannel());
	}
	protected <T> ListenableFuture<T> asyncCall(Channel<Envelope> requestChannel, 
			Envelope envelope,Class<T> reponseType) {
		checkNotNull(requestChannel,"requestChannel is null");
		checkNotNull(envelope,"envelope is null");
		if (null == envelope.sendTime) {
			envelope.sendTime = new Date();
		}
		ListenableFuture<T> future;
		if(reponseType == null) {
			/** 没有指定返回值， 则立即返回 */
			future = Futures.immediateFuture(null);
		}else {
			future = new InvocationContext<>(envelope, reponseType).future;
			if (taskTimeoutMillis > 0) {
				future = Futures.withTimeout(future, taskTimeoutMillis, TimeUnit.MILLISECONDS, scheduledExecutor);
			}
		}
		publisher.publish(requestChannel, envelope);
		return future;
	}

	/**
	 * [异步]调用指定的远程方法
	 * @param <T>
	 * @param requestChannel 调用目标频道
	 * @param methodName 调用方法名
	 * @param args 调用参数
	 * @param reponseType 返回类型，如果为 {@code null} ，立即返回
	 * @return 表示异步计算的结果的异步调用对象
	 */
	public <T> ListenableFuture<T> asyncCall(String requestChannel, String methodName, Map<String,Object> args,
			Class<T> reponseType) {
		checkArgument(!isNullOrEmpty(requestChannel),"requestChannel is null or empty");
		checkArgument(!isNullOrEmpty(methodName),"methodName is null or empty");
		Envelope envelope = createEnvelope(methodName, args, reponseType);
		return asyncCall(new Channel<>(requestChannel, Envelope.class), envelope, reponseType);
	}
	/**
	 * [异步]调用指定的远程方法
	 * @param <T>
	 * @param requestChannel 调用目标频道
	 * @param methodName 调用方法名
	 * @param args 调用参数
	 * @param reponseType 返回类型，如果未指定则视为void类型，立即返回
	 * @param callback 异步回调方法
	 */
	public <T> void asyncCall(String requestChannel, String methodName, Map<String,Object> args,
			Class<T> reponseType,FutureCallback<? super T> callback) {
		Futures.addCallback(asyncCall(requestChannel,methodName,args,reponseType), callback);
	}
	/**
	 * [同步]调用指定的远程方法
	 * @param <T>
	 * @param requestChannel 调用目标频道
	 * @param methodName 调用方法名
	 * @param args 调用参数，无参数置为{@code null}
	 * @param reponseType 返回类型，如果未指定则视为void类型，立即返回
	 * @return 返回调用结果
	 */
	public <T> T syncCall(String requestChannel, String methodName, Map<String,Object> args,
			Class<T> reponseType) {
		try {
			return Futures.getUnchecked(asyncCall(requestChannel,methodName,args,reponseType));
		} catch (UncheckedExecutionException e) {
			Throwables.throwIfUnchecked(e.getCause());
			throw new RuntimeException(e.getCause());
		}
	}

	/**
	 * 调用指定的无返回值远程方法
	 * @param requestChannel 调用目标频道
	 * @param methodName 调用方法名
	 * @param args 调用参数，无参数置为{@code null}
	 * @param immediately 为{@code true}立即返回,否则同步等待远端响应后返回
	 * @see #asyncCall(String, String, Map, Class)
	 */
	public void onewayCall(String requestChannel, String methodName, Map<String,Object> args, boolean immediately) {
		if(immediately) {
			asyncCall(requestChannel,methodName,args, null);
		}else {
			Futures.getUnchecked(asyncCall(requestChannel,methodName,args,void.class));
		}
	}

	private <T> Envelope createEnvelope(String methodName, Map<String, Object> args, Class<?> reponseType) {
		Envelope envelope = new Envelope();
		envelope.setAck(getAckChannel());
		JSONArray array = new JSONArray();
		/**
		 * 生成消息体， 参见 org.apache.thrift.protocol.TMessage
		 */
		array.add(1);/** VERSION */
		array.add(methodName);/** name */
		array.add(isVoid(reponseType)?4:1);/** org.apache.thrift.protocol.TMessageType.CALL,ONEWAY */
		array.add(SEQID.incrementAndGet());/** TMessage.seqid */
		array.add(firstNonNull(args, Collections.emptyMap()));
		envelope.bodyArray(array);
		return envelope.buildBody();
	}

	protected String getGeneratedAck() {
		if (null == autoAck) {
			synchronized (this) {
				if (null == autoAck) {
					autoAck = "mqrpc_" + System.nanoTime();
				}
			}
		}
		return autoAck;
	}

	private String getAckChannel() {
		if (null != ackSupplier) {
			return ackSupplier.get();
		}
		return getGeneratedAck();
	}

	@Override
	public void onSubscribe(Envelope envelope) throws SmqUnsubscribeException {
		/** 忽略请求发起时间超时的请求 */
		if (null != envelope.sendTime
				&& Math.abs(System.currentTimeMillis() - envelope.sendTime.getTime()) > taskTimeoutMillis) {
			return;
		}
		InvocationContext<?> ctx;
		try {
			/** 通过请求序列号找到对应的 InvocationContext 对象 */
			int seqid = envelope.seqid();
			ctx = invocationContexts.get(seqid);
			if (null == ctx) {
				/** 忽略请求序列号无效的消息 */
				SimpleLog.log("INVALID SEQID {}", seqid);
				return;
			}
		} catch (JSONException e) {
			SimpleLog.log(e.getMessage());
			return;
		}
		try {
			/** 检查请求方法名是否匹配 */
			if (envelope.name().equals(ctx.name())) {
				JSONObject content = envelope.content();
				if (content.containsKey(FIELD_SUCCESS)) {
					ctx.sucess(content.getObject(FIELD_SUCCESS, ctx.reponseType));
				} else if (content.containsKey(FIELD_EXCEPTION)) {
					throw exceptionType.getConstructor(ResponseException.class).newInstance(content.getObject(FIELD_EXCEPTION, ResponseException.class));
				} else if (content.containsKey(FIELD_MESSAGE)) {
					throw new TApplicationException(content.getIntValue(FIELD_TYPE),content.getString(FIELD_MESSAGE));
				} else if(isVoid(exceptionType) || content.isEmpty()){
					ctx.sucess(null);
				}else {
					throw new InvalidResponseException(String.format("NOT FOUND REQUIRED FIELD: '%s' OR '%s'",
							FIELD_SUCCESS, FIELD_EXCEPTION));
				}
			} else {
				throw new InvalidResponseException(String.format("MISMATCH request method name %s VS %s",
						envelope.name(), ctx.name()));
			}
		} catch (JSONException e) {
			ctx.exception(new InvalidResponseException(e.getMessage(), e));
		}catch (Exception e) {
			ctx.exception(e);
		}
	}

	private static boolean isVoid(Class<?> clazz) {
		return null == clazz ? true : Primitives.unwrap(clazz).equals(void.class);
	}

	public static Builder builder() {
		return new Builder();
	}
	/**
	 * 调用上下文
	 * @author guyadong
	 *
	 * @param <T>
	 */
	private static class InvocationContext<T> extends Envelope implements Runnable{
		final Class<T> reponseType;
		final SettableFuture<T> future;
		
		InvocationContext(Envelope envelope, Class<T> reponseType) {
			bodyArray((JSONArray) envelope.bodyArray().clone());
			this.reponseType = reponseType;
			this.future = SettableFuture.create();
			/** 增加侦听器，执行结束时从 invocationContexts 中删除当前对象 */
			this.future.addListener(this, MoreExecutors.directExecutor());
			invocationContexts.put(seqid(), this);
		}

		@SuppressWarnings({ "unchecked" })
		public void sucess(Object sucess) {
			this.future.set((T)sucess);
		}
		public void exception(Throwable throwable) {
			this.future.setException(throwable);
		}
		@Override
		public void run() {
			invocationContexts.remove(seqid());
			
		}
	}
	public static class Builder{
		IMessageQueueFactory messageQueueFactory;
		long taskTimeoutMillis = DEFAULT_TASK_TIMEOUT_MILLIS;
		ScheduledExecutorService scheduledExecutor;
		Class<? extends Exception> exceptionType;
		Supplier<String> ackSupplier;
		Builder() {}
		public Builder messageQueueFactory(IMessageQueueFactory messageQueueFactory) {
			this.messageQueueFactory = messageQueueFactory;
			return this;
		}
		public Builder taskTimeoutMillis(long taskTimeoutMillis) {
			this.taskTimeoutMillis = taskTimeoutMillis;
			return this;
		}
		public Builder taskTimeout(TimeUnit timeUnit, long duration) {
			this.taskTimeoutMillis = timeUnit.toMillis(duration);
			return this;
		}
		public Builder scheduledExecutor(ScheduledExecutorService scheduledExecutor) {
			this.scheduledExecutor = scheduledExecutor;
			return this;
		}
		public Builder exceptionType(Class<? extends Exception> exceptionType) {
			this.exceptionType = exceptionType;
			return this;
		}
		public Builder ackSupplier(Supplier<String> ackSupplier) {
			this.ackSupplier = ackSupplier;
			return this;
		}
		public Builder ackChannel(String ackChannel) {
			return ackSupplier(Suppliers.ofInstance(ackChannel));
		}
		
		public MqInvoker build() {
			return new MqInvoker(this);
		}
	}
}
